package com.hps.demo.msg;

import com.alibaba.fastjson.JSON;
import com.hps.demo.entity.Order;
import com.hps.demo.service.IOrderService;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * @author heps
 * @since 2020/11/13 15:36
 */
@Slf4j
@Component
public class OrderTopicListener implements InitializingBean, DisposableBean {

  private static final BlockingQueue<Order> ORDER_QUEUE = new LinkedBlockingQueue<>();
  
  private volatile Boolean running = true;

  @Resource
  private IOrderService orderService;

  @Resource
  private Executor bizExecutor;

  @KafkaListener(topics = "orderTopic", groupId = "orderTopicGroup")
  public void listen(ConsumerRecord<String, String> msg) {
    Order order = JSON.parseObject(msg.value(), Order.class);
    ORDER_QUEUE.offer(order);
    if (ORDER_QUEUE.size() > 3000) {
      try {
        TimeUnit.SECONDS.sleep(1);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }

  public void saveData() {
    List<Order> orders = new ArrayList<>();
    while (running) {
      try {
        Order order = ORDER_QUEUE.poll(10, TimeUnit.MILLISECONDS);
        if (Objects.nonNull(order)) {
          orders.add(order);
        } else {
          TimeUnit.MILLISECONDS.sleep(500);
        }
        if (orders.size() > 0 && (Objects.isNull(order) || orders.size() % 100 == 0)) {
          log.info("order queue size: {}", ORDER_QUEUE.size());
          orderService.saveBatch(orders);
          orders.clear();
        }
      } catch (InterruptedException e) {
        log.error(e.getMessage(), e);
      }

    }
    while (ORDER_QUEUE.size() > 0) {
      orders.add(ORDER_QUEUE.poll());
    }
    if (orders.size() > 0) {
      orderService.saveBatch(orders);
    }
    try {
      TimeUnit.SECONDS.sleep(5);
      if (ORDER_QUEUE.size() > 0) {
        log.info("the order queue is not empty, good consumer cannot stop");
        startConsumer();
      } else {
        log.info("order consumer stopped...");
      }
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }

  public void setRunning(boolean running) {
    this.running = running;
  }

  public void startConsumer() {
    if (!running) {
      synchronized (running) {
        if (!running) {
          running = true;
          bizExecutor.execute(this::saveData);
        }
      }
    }
  }

  @Override
  public void afterPropertiesSet() throws Exception {
    bizExecutor.execute(this::saveData);
    log.info("order topic listener started...");
  }

  @Override
  public void destroy() throws Exception {
    if (running) {
      running = false;
      TimeUnit.SECONDS.sleep(5);
    }
  }
}
